package zhanglu.family.socket.textline.busclient;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import net.sf.json.JSONObject;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import zhanglu.family.socket.bean.ErrMsg;
import zhanglu.family.socket.bean.busbean.BusHeadBean;
import zhanglu.family.socket.bean.busbean.BusMsgBean;
import zhanglu.family.socket.bean.busbean.BusPostAnnotation;
import zhanglu.family.socket.util.JsonUtil;

/**
 * 实例化客户端端的管理类
 * 
 * @author zhanglu
 * @date 2016年12月16日
 */
public class BusTCPClient {

    private static Logger        logger   = LoggerFactory.getLogger(BusTCPClient.class);

    private NioSocketConnector   connector;                                             // 总进程

    private ConnectFuture        connectFuture;                                         // 客户端连接

    private IoHandlerAdapter     handlerAction;                                         // 业务处理类

    private int                  idleTime = 10;                                         // 连接超时时间秒,默认10秒

    private String               ip;                                                    // 服务端IP

    private int                  port;                                                  // 服务端端口

    private int                  reconnectionTime;                                      // 客户端重连间隔时间秒

    private String               charset;                                               // 编码格式

    public Map<Long, BusMsgBean> synMsgMap;                                             // 用于做消息处理

    public long                  seqId    = -1;                                         // 用于同步消息流水号,从0开始自减1，所以所有异步消息流水号不准为同样的数字

    private long                 sessionSequence;                                       // 此客户端的身份

    /**
     * 实例化对象
     * 默认超时60秒、重连时间间隔5秒、编码UTF-8，可用方法重新设置
     * 
     * @param ip
     *            服务端IP
     * @param port
     *            服务端端口
     */
    public BusTCPClient(String ip, int port) {
        this.idleTime = 60;
        this.ip = ip;
        this.port = port;
        this.reconnectionTime = 5;
        this.charset = "UTF-8";
        this.synMsgMap = new ConcurrentHashMap<Long, BusMsgBean>();
    }

    /**
     * 设置超时时间
     * 
     * @param idleTime
     *            超时时间（秒）
     */
    public void setIdleTime(int idleTime) {
        this.idleTime = idleTime;
    }

    /**
     * 设置重连时间
     * 
     * @param reconnectionTime
     *            重连时间（秒）
     */
    public void setReconnectionTime(int reconnectionTime) {
        this.reconnectionTime = reconnectionTime;
    }

    /**
     * 设置编码格式
     * 
     * @param charset
     *            编码格式
     */
    public void setCharset(String charset) {
        this.charset = charset;
    }

    /**
     * 启动
     * 
     * @param handlerAction
     *            业务处理类
     * @param sessionSequence
     *            客户端身份标识,每个客户端身份标识唯一
     */
    public void startClient(IoHandlerAdapter handlerAction, long sessionSequence) {
        this.handlerAction = handlerAction;
        this.sessionSequence = sessionSequence;
        reconnection();
    }

    /**
     * 发送异步消息
     * 
     * @param msg
     *            消息体
     */
    private void post(BusMsgBean msg) {
        try {

            if (this.connectFuture != null) {
                JSONObject json = JSONObject.fromObject(msg);
                this.connectFuture.getSession().write(json.toString());
            } else {
                logger.debug("没有连接到总线");
            }
        } catch (Exception e) {
            logger.error("连接错误", e);
        }

    }

    /**
     * 发送同步消息
     * 
     * @param msg
     *            消息对象
     * @param secondTimeout
     *            超时时间(秒)
     * @return
     */
    private BusMsgBean send(BusMsgBean msg, int secondTimeout) {

        long seq = seqId;// 这里接一下，下面马上开始--，防止并发(这里防止单一客户端，多客户端情况不会并发)

        this.seqId--;// 流水号用过之后立即-1
        if (this.seqId <= Long.MIN_VALUE) {// 如果超出限制，归-1
            this.seqId = -1;
        }

        BusMsgBean msgResult = new BusMsgBean();

        try {

            if (this.connectFuture == null || !connectFuture.isConnected()) {// 没有连接到总线
                msgResult.setErrMsg(ErrMsg.NOCONNECTION);
                return msgResult;
            }

            BusMsgBean wait = new BusMsgBean();
            wait.setErrMsg(ErrMsg.WAIT);
            this.synMsgMap.put(seq, wait);// 这里将同步流水号放入。value为null，后面人为阻塞后拿值,根据拿到的类中一些判断逻辑继续进行

            msg.getHead().setSeqId(seq);
            msg.setErrMsg(ErrMsg.WAIT);

            this.connectFuture.getSession().write(JsonUtil.ObToJsonStr(msg));

            int interval = 100;// 定时间隔（毫秒）
            int tick = 0;// 计数器，定时查询反馈消息
            while (tick <= secondTimeout * 1000) {// 人为阻塞获取消息体
                BusMsgBean msgReceive = this.synMsgMap.get(seq);

                if (msgReceive.getErrMsg() == ErrMsg.NOTARGETSESSION) {// 对方没有上线
                    msgResult.setErrMsg(ErrMsg.NOTARGETSESSION);

                    this.synMsgMap.remove(seq);

                    break;
                }

                if (msgReceive.getErrMsg() == ErrMsg.NONE) {// 如果接受到了数据，则返回
                    msgResult.setJsonMsg(msgReceive.getJsonMsg());
                    msgResult.setErrMsg(ErrMsg.NONE);

                    this.synMsgMap.remove(seq);

                    break;
                }

                tick += interval;

                if (tick > secondTimeout * 1000) {// 访问超时，清除同步消息。
                    msgResult.setErrMsg(ErrMsg.CLIENTTIMEOUT);
                    this.synMsgMap.remove(seq);

                    break;
                }

                Thread.sleep(interval);
            }
        } catch (Exception e) {
            logger.error("连接错误", e);
        }

        return msgResult;
    }

    /**
     * 关闭连接
     */
    public void closeSession() {
        if (this.connectFuture != null && this.connectFuture.getSession() != null) {
            this.connectFuture.getSession().close(true);
        }
    }

    /**
     * 杀掉整个连接的进程
     */
    public void dispose() {
        if (this.connector != null) {
            this.connector.dispose();
            this.connector = null;
            logger.info("进程已经杀死");
        }
    }

    /**
     * 重连
     * 
     * @param sessionSequence
     *            客户端身份标识,每个客户端身份标识唯一
     */
    public void reconnection() {

        this.connectFuture = null;// 先让连接等于空，然后进行重连

        if (this.connector == null) {

            this.connector = new NioSocketConnector();
            // 创建接受数据的过滤器
            DefaultIoFilterChainBuilder chain = this.connector.getFilterChain();
            // 设定这个过滤器将一行一行(/r/n)的读取数据
            chain.addLast("myChin", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName(this.charset))));// 文本处理

            // 客户端的消息处理器：一个SamplMinaServerHander对象
            this.connector.setHandler(this.handlerAction);
            // set connect timeout
            this.connector.setConnectTimeoutMillis(this.idleTime * 1000);
        }

        while (this.connectFuture == null || !connectFuture.isConnected()) {
            try {

                this.connectFuture = connector.connect(new InetSocketAddress(this.ip, this.port));
                this.connectFuture.awaitUninterruptibly();// 这一行阻塞，等连接上了继续向下走，否则正在连接的过程中直接发数据会报错

                BusMsgBean msg = new BusMsgBean();
                BusHeadBean bean = new BusHeadBean();
                bean.setHeadId(-1L);
                bean.setSessionSequence(this.sessionSequence);
                msg.setHead(bean);

                this.connectFuture.getSession().write(JsonUtil.ObToJsonStr(msg));

                logger.info(this.sessionSequence + "客户端连接成功,服务端ip为" + this.ip + ",服务端端口为" + this.port);
                break;

            } catch (Exception e) {
                try {
                    this.connectFuture = null;
                    logger.info("客户端连接失败进行重连,服务端ip为" + this.ip + ",服务端端口为" + this.port);
                    Thread.sleep(this.reconnectionTime * 1000);
                    continue;
                } catch (InterruptedException e1) {
                    // e1.printStackTrace();
                    continue;
                }
            }
        }
    }

    /**
     * 异步发送消息
     * 
     * @param headId
     *            协议号。标示具体某条协议
     * @param seqId
     *            消息流水号，标示消息唯一
     * @param targetSession
     *            目标人
     * @param bean
     *            消息体
     */
    public void post(long headId, long seqId, long targetSession, Object bean) {
        BusMsgBean msg = new BusMsgBean();// 消息体
        BusHeadBean head = new BusHeadBean();// 消息头

        head.setHeadId(headId);
        head.setSeqId(seqId);
        head.setSessionSequence(this.sessionSequence);
        head.setTargetSession(targetSession);

        msg.setJsonMsg(JsonUtil.ObToJsonStr(bean));
        msg.setHead(head);// 这个必须加。需用通用头协议来判定，只需要将自己的bean直接set到这个字段内即可
        post(msg);
    }

    /**
     * 异步发送消息(注解)
     * 
     * @param t
     *            带有注解的实体类,即发送类
     * @param seqId
     *            消息流水号
     * @param bean
     *            消息文本
     */
    public void post(Class<?> t, long seqId, Object bean) {
        BusPostAnnotation post = t.getAnnotation(BusPostAnnotation.class);

        post(post.headId(), seqId, post.targetSession(), bean);
    }

    /**
     * 同步发送消息
     * 
     * @param targetSession
     *            目标人
     * @param secondTimeout
     *            超时时间秒
     * @param bean
     *            消息文本
     * @return
     */
    public BusMsgBean send(long targetSession, int secondTimeout, Object bean) {
        BusMsgBean msg = new BusMsgBean();// 消息体
        BusHeadBean head = new BusHeadBean();// 消息头

        head.setSessionSequence(this.sessionSequence);
        head.setTargetSession(targetSession);

        msg.setJsonMsg(JsonUtil.ObToJsonStr(bean));
        msg.setHead(head);// 这个必须加。需用通用头协议来判定，只需要将自己的bean直接set到这个字段内即可

        return send(msg, secondTimeout);
    }
}
